package com.heyi.nettylong.client;

import com.heyi.nettylong.client.config.ClientProperties;
import com.heyi.nettylong.client.handler.HeartHandler;
import com.heyi.nettylong.client.handler.LoginHandler;
import com.heyi.nettylong.common.MessageHandler;
import com.heyi.nettylong.common.MessageRepository;
import com.heyi.nettylong.common.protobuf.Command;
import com.heyi.nettylong.common.protobuf.Message;

import com.heyi.nettylong.common.util.MessageUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutor;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Promise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * netty客户端
 * @author lenovo
 *
 */
public class NettyClient extends Thread {
	private final Logger log = LoggerFactory.getLogger(this.getClass());

	public ClientProperties clientProperties;

	private final EventLoopGroup loop = new NioEventLoopGroup();
	public static Channel clientChannel;
	public static int connectAuthFail = 0;

	public NettyClient(ClientProperties clientProperties){
		this.clientProperties = clientProperties;
	}

	@Override
	public void run()  {
		try {
			doConnect(new Bootstrap(), loop);
		}catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * netty client 连接，连接失败5秒后重试连接
	 */
	public Bootstrap doConnect(Bootstrap bootstrap, EventLoopGroup eventLoopGroup) {
		log.info("-----开始连接netty服务端：{}_{}", clientProperties.getServerHost(), clientProperties.getServerPort());
		try {
			if (bootstrap != null) {
				bootstrap.group(eventLoopGroup);
				bootstrap.channel(NioSocketChannel.class);
				bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
				bootstrap.handler(new ChannelInitializer<SocketChannel>() {
					@Override
					public void initChannel(SocketChannel ch) throws Exception {
						ChannelPipeline p = ch.pipeline();

						p.addLast("idleStateHandler", new IdleStateHandler(clientProperties.getHeartSeconds(), 0
								, 0, TimeUnit.SECONDS));

						p.addLast(new ProtobufVarint32FrameDecoder());
						p.addLast(new ProtobufDecoder(Message.MessageBase.getDefaultInstance()));

						p.addLast(new ProtobufVarint32LengthFieldPrepender());
						p.addLast(new ProtobufEncoder());

						p.addLast("clientHandler", new LoginHandler(NettyClient.this));
						p.addLast("idleTimeoutHandler", new HeartHandler(clientProperties));

					}
				});
				bootstrap.remoteAddress(clientProperties.getServerHost(), clientProperties.getServerPort());
				ChannelFuture f = bootstrap.connect().addListener((ChannelFuture futureListener)->{
					final EventLoop eventLoop = futureListener.channel().eventLoop();
					if (!futureListener.isSuccess()) {
						log.info("连接服务器失败，5s后重新尝试连接！");
						futureListener.channel().eventLoop().schedule(() -> doConnect(new Bootstrap(), eventLoop), 5, TimeUnit.SECONDS);
					}
				});
				f.channel().closeFuture().sync();
				//eventLoopGroup.shutdownGracefully();
			}
		} catch (InterruptedException e) {
			log.error("-----连接netty服务端失败："+e.getMessage());
			e.printStackTrace();
		}
		return bootstrap;
	}

	/**
	 * 发送消息
	 * @param text 消息内容
	 * @return
	 * @throws Exception
	 */
	public ChannelFuture sendMessage(String text) throws Exception {
		// 判断是否活跃
		if(clientChannel==null || !clientChannel.isActive()){
			throw new Exception("已与服务端断开连接");
		}
		return clientChannel.writeAndFlush(MessageUtil.build(clientProperties.getClientId(), Command.CommandType.PUSH_DATA, text, clientProperties.getToken()));
	}

	/**
	 * 发送消息并得到返回的消息(阻塞方法，谨慎使用)
	 * @param text 消息内容
	 * @param timeout 等待客户端返回消息的时间，单位：秒
	 * @return 客户返回的消息
	 * @throws Exception
	 */
	public synchronized String sendMessageAndReturn(String text, long timeout) throws Exception {
		// 判断是否活跃
		if(clientChannel==null || !clientChannel.isActive()){
			throw new Exception("已与服务端断开连接");
		}
		int msgId = MessageRepository.buildSequence();
		// 构造线程池
		EventExecutor executor = new DefaultEventExecutor();
		// 创建 DefaultPromise 实例
		Promise promise = new DefaultPromise(executor);
		MessageRepository.addListener(msgId, (String content) ->
				promise.setSuccess(content));

		clientChannel.writeAndFlush(MessageUtil.build(msgId, clientProperties.getClientId(), Command.CommandType.PUSH_DATA, text, clientProperties.getToken()));

		String msg = null;
		try{
			msg = promise.get(timeout, TimeUnit.SECONDS).toString();
		}catch (Exception ex){
			MessageRepository.removeListener(msgId);
			throw new Exception("客户端返回消息超时");
		}
		return msg;
	}
}
